package com.bj58.test

import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.sql.{Row, SparkSession}
import util.{Compute, RoiUdfUtil, SparkReadUtil}
import org.apache.spark.sql.functions.lit
import org.apache.spark.sql.functions.col

import scala.collection.mutable.ArrayBuffer
import scala.util.Try

/**
  * Created by 6v on 2018/11/5.
  */
object RoiSearchRead {

  val URL = "/home/hdp_lbg_ectech/resultdata/adsp/dp/ods/roi/search/"
  val SPLIT_WORD = '\u0001'
  val SPLIT_LINE = '\n'

  val COLUMN_SEQ = Seq("sid", "gtid", "sessionid", "cookieid", "sloc1", "sloc2", "sloc3", "sloc4", "scate1", "scate2"
    , "scate3", "scate4", "scate5", "url", "pagetype", "ip", "imei", "xforward", "ua", "pf"
    , "userid", "outkword", "inkword", "utmsrc", "spm", "params", "validst", "hastrack", "infouserid", "infousertype"
    , "infoid", "infotype", "adtype", "stime", "isextend", "pageno", "pos", "clickid", "clicktime", "clicktag"
    , "clicktype", "sourcepage", "seq", "staytime", "ab")

  val COLUMN_OUTPUT = Seq("cookieid", "sloc1", "sloc2", "sloc3", "scate1", "scate2", "scate3"
    , "ip","imei",   "ua", "userid"
    , "infoid", "infotype",  "clicktime", "clicktag","stime", "pageno", "pos"
    , "sourcepage", "ab")

  val OUTPUT =Seq("cookieid", "sloc1", "sloc2", "sloc3", "scate1", "scate2", "scate3" , "ip","imei",   "ua",
    "userid", "infoid", "infotype",  "clicktime", "clicktag","pos_type", "abtest", "user_type" , "create_time", "page_id"
    , "exp_n", "operation_type" , "platform")



  def main(args: Array[String]): Unit = {

    val spark = SparkSession
      .builder()
      .appName("RoiClickRead")
      .master("local[4]")
      .getOrCreate()


    val path = "C:\\Users\\lenovo\\Desktop\\2.txt"
    var data = SparkReadUtil.readFromFile(spark, path, SPLIT_LINE, SPLIT_WORD, COLUMN_SEQ, COLUMN_OUTPUT).where("validst = 1")
    data = data
      .withColumn("user_type",RoiUdfUtil.getUserTypeFromUserId(data("userid")))
      .withColumnRenamed("stime", "create_time")
      .withColumnRenamed("sourcepage", "pos_type")//点击来源页面类型，取值为：{-1:未知，1:列表页，2:详情页}
      .withColumnRenamed("pageno", "page_id")
      .withColumnRenamed("pos", "exp_n")
      .withColumn("operation_type",lit("1"))
      .withColumn("platform",RoiUdfUtil.getPlatform(data("ua")))
//      .withColumn("info_app_id",RoiUdfUtil.getInfoApp(data("infoid")))
      .withColumnRenamed("ab", "abtest")


    val pathClick = "C:\\Users\\lenovo\\Desktop\\1 (2).txt"
    var dataClick = SparkReadUtil.readFromFile(spark, pathClick, SPLIT_LINE, SPLIT_WORD, RoiClickRead.COLUMN_SEQ, RoiClickRead.COLUMN_OUTPUT).where("validst = 1")
    dataClick = dataClick
      .withColumn("user_type",RoiUdfUtil.getUserTypeFromUserId(col("userid")))
      .withColumn("create_time", col("clicktime"))
      .withColumnRenamed("sourcepage", "pos_type")//点击来源页面类型，取值为：{-1:未知，1:列表页，2:详情页}
      .withColumn("page_id", lit("0"))
      .withColumn("exp_n", lit("0"))
      .withColumn("operation_type",lit("2"))
      .withColumn("platform",RoiUdfUtil.getPlatform(col("ua")))
//      .withColumn("info_app_id",RoiUdfUtil.getInfoApp(col("infoid")))
//    dataClick.show(5)

//println(data.schema)
//    println(dataClick.schema)
    val unionData = data.select(OUTPUT.head,OUTPUT.tail:_*).union(dataClick).select(OUTPUT.head,OUTPUT.tail:_*)
    unionData.show(5)

    val resRDD = unionData.rdd.map(row=>{
      val imei = row.getAs[String]("imei")
      (imei, ArrayBuffer(row))
    }).reduceByKey((left, right) => {
      left.++=(right)
    }).flatMap(tup => {
      val rows = new ArrayBuffer[Row]
      val imei = tup._1

      val arr = tup._2
      val action = arr.filter(_.getAs[String]("operation_type") != "1")
      val view = arr.filter(_.getAs[String]("operation_type") == "1").sortBy(row => Try(row.getAs[String]("create_time").toLong).getOrElse(0L))
      println("====="+action.length)
      println("====="+view.length)
      if (view.nonEmpty) {
        val timeArr = view.map(row => Try(row.getAs[String]("create_time").toLong).getOrElse(0L)).toArray
        //          scala.util.Sorting.quickSort(timeArr)
        action.map(row => {
          val index = Compute.findFirstLarger(timeArr, Try(row.getAs[String]("create_time").toLong).getOrElse(0L))
          val i = if (index > 0) index - 1 else 0
          val exp_n = view(i).getAs[String]("exp_n")
          val page_id = view(i).getAs[String]("page_id")

          var res = new ArrayBuffer[Any]
          println("￥￥￥￥￥￥￥￥￥￥￥￥￥￥￥￥￥"+exp_n)
          val cookieid = row.getAs[String]("cookieid")
          val sloc1 = row.getAs[String]("sloc1")
          val sloc2 = row.getAs[String]("sloc2")
          val sloc3 = row.getAs[String]("sloc3")
          val scate1 = row.getAs[String]("scate1")
          val scate2 = row.getAs[String]("scate2")
          val scate3 = row.getAs[String]("scate3")
          val ip = row.getAs[String]("ip")
          val ua = row.getAs[String]("ua")
          val userid = row.getAs[String]("userid")
          val infoid = row.getAs[String]("infoid")
          val infotype = row.getAs[String]("infotype")
          val clicktime = row.getAs[String]("clicktime")
          val clicktag = row.getAs[String]("clicktag")
          val pos_type = row.getAs[String]("pos_type")
          val abtest = row.getAs[String]("abtest")
          val user_type = row.getAs[Int]("user_type")
          val create_time = row.getAs[String]("create_time")
//          val exp_n = row.getAs[String]("exp_n")
          val operation_type = row.getAs[String]("operation_type")
          val platform = row.getAs[Int]("platform")

          res += cookieid
          res += sloc1
          res += sloc2
          res += sloc3
          res += scate1
          res += scate2
          res += scate3
          res += ip
          res += imei
          res += ua
          res += userid
          res += infoid
          res += infotype
          res += clicktime
          res += clicktag
          res += pos_type
          res += abtest
          res += user_type
          res += create_time
          res += page_id
          res += exp_n
          res += operation_type
          res += platform
          val newRow = Row.fromSeq(res)
          rows += newRow
        })
      }
      println("==============="+rows.length)
      rows
    })

    println("#####"+unionData.schema)

    var res = spark.createDataFrame(resRDD, dataClick.schema)
    res.show(10)















    //    println("count:"+data.count())


  }

}
